package spring.cloud.tasks.tasks_assign_manager;


import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.springframework.context.annotation.Configuration;
import spring.cloud.tasks.common.node.ExecutorNodePath;
import spring.cloud.tasks.common.node.TaskNodePath;
import spring.cloud.tasks.common.zookeeper.ZooKeeper;
import spring.cloud.tasks.common.zookeeper.ZooKeeperRepository;
import spring.cloud.tasks.tasks_assign_manager.database.entity.NameSpaceZooKeeperClusterMappingDto;
import spring.cloud.tasks.tasks_assign_manager.database.entity.ZooKeeperClusterDto;
import spring.cloud.tasks.tasks_assign_manager.database.service.NameSpaceDtoService;
import spring.cloud.tasks.tasks_assign_manager.database.service.NameSpaceZooKeeperClusterMappingDtoService;
import spring.cloud.tasks.tasks_assign_manager.database.service.ZooKeeperClusterDtoService;
import spring.cloud.tasks.tasks_assign_manager.domain.NameSpace;
import spring.cloud.tasks.tasks_assign_manager.domain.NameSpaceZooKeeperClient;
import spring.cloud.tasks.tasks_assign_manager.domain.ZooKeeperCluster;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

@Configuration
@Slf4j
public class BootStrap {


    public static final ConcurrentHashMap<String, Object> nameSpaceToLockMap = new ConcurrentHashMap<String, Object>();
    public static final ConcurrentHashMap<String, NameSpaceZooKeeperClient> nameSpaceTonameSpaceZooKeeperClientMap = new ConcurrentHashMap<String, NameSpaceZooKeeperClient>();
    private static final Set<String> restrictComputeZooKeeperClusterKeyList = Sets.newHashSet();
    private static final ConcurrentHashMap<String, TasksManager> nameSpaceToTaskManagerMap = new ConcurrentHashMap<String /** nns **/, TasksManager>();
    //
    public static LinkedHashMap<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap = new LinkedHashMap<>();
    private final ReentrantLock lock = new ReentrantLock();
    @Resource
    private ZooKeeperClusterDtoService zooKeeperClusterDtoService;
    @Resource
    private NameSpaceDtoService nameSpaceDtoService;
    @Resource
    private NameSpaceZooKeeperClusterMappingDtoService nameSpaceZooKeeperClusterMappingDtoService;
    @Resource
    private ZooKeeperRepository zooKeeperRepository;
    private String tasksManagerId;
    private List<String> onlinenameSpaceList = new ArrayList<String>();
    private Timer refreshTimer = null;
    private Timer refreshIfNecessaryTimer = null;

    public static Object getNameSpaceLock(String nameSpace) {
        Object lock = nameSpaceToLockMap.get(nameSpace);
        if (lock == null) {
            lock = new Object();
            Object pre = nameSpaceToLockMap.putIfAbsent(nameSpace, lock);
            if (pre != null) {
                lock = pre;
            }
        }
        return lock;
    }

    public void start() throws Exception {
        init();
    }

    public void stop() throws Exception {
        destroy();
    }

    public void init() {
        refresh();
        startRefreshTimer();
        startRefreshIfNecessaryTimer();//这个时间短
    }

    private void startRefreshTimer() {
        refreshTimer = new Timer("refresh-timer", true);
        // 每隔5分钟刷新一次
        refreshTimer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                try {
                    Runnable runnable = (new Runnable() {
                        @Override
                        public void run() {
                            try {
                                boolean tryLock = lock.tryLock(10, TimeUnit.SECONDS);
                                if (tryLock) {
                                    lock.unlock();
                                    refresh();//避免等的时间太长
                                }
                            } catch (Exception e) {
                                log.error(e.getMessage(), e);
                            }
                        }
                    });
                    Thread thread = new Thread(runnable);
                    thread.start();
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }, 1000L * 60 * 5, 1000L * 60 * 5);
    }

    private void startRefreshIfNecessaryTimer() {
        refreshIfNecessaryTimer = new Timer("refresh-if-necessary-timer", true);
        refreshIfNecessaryTimer.schedule(new TimerTask() {
            private String lastUuid = null;

            @Override
            public void run() {
                try {
                    String uuid = null;//systemConfigurationService.getValueDirectly(SystemConfigProperties.REFRESH_REGISTRY_CENTER_UUID);
                    if (uuid == null) {
                        return;
                    } else if (!uuid.equals(lastUuid)) {
                        lastUuid = uuid;
                        Runnable runnable = (new Runnable() {
                            @Override
                            public void run() {
                                try {
                                    refresh();
                                } catch (Exception e) {
                                    log.error(e.getMessage(), e);
                                }
                            }
                        });
                        Thread thread = new Thread(runnable);
                        thread.start();
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }, 1000 * 5, 1000 * 5);
    }

    @PreDestroy
    public void destroy() {
        Iterator<Map.Entry<String, ZooKeeperCluster>> iterator = zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
        while (iterator.hasNext()) {
            closeZooKeeperCluster(iterator.next().getValue());
        }

        if (refreshTimer != null) {
            refreshTimer.cancel();
        }
        if (refreshIfNecessaryTimer != null) {
            refreshIfNecessaryTimer.cancel();
        }

    }

    private synchronized void refresh() {
        try {
            Map<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap = refreshRestrictComputeZooKeeperClusterKeyList();
            if (restrictComputeZooKeeperClusterKeyList.isEmpty()) {
                //如果存在之前的数据则需要清理
                Iterator<Map.Entry<String, ZooKeeperCluster>> iterator = zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
                while (iterator.hasNext()) {
                    closeZooKeeperCluster(iterator.next().getValue());
                }
                log.warn("根据Console的集群ID:" + "" + ",找不到配置可以参与Sharding和Dashboard计算的zk集群");
                return;
            }
            refreshZooKeeperClusterAndnameSpace(zooKeeperClusterKeyToZooKeeperClusterMap);
            refreshnameSpaceToTaskManagerMap();
        } catch (Exception e) {
            log.error("refresh RegCenter error", e);
        }
    }

    private Map<String, ZooKeeperCluster> refreshRestrictComputeZooKeeperClusterKeyList() throws RuntimeException {
        //DB最新的
        Map<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap = getZooKeeperClusterKeyToZooKeeperClusterMap();
        List<String> zooKeeperClusterKeyList = getZooKeeperClusterKeyList(zooKeeperClusterKeyToZooKeeperClusterMap);
        // clear 当前可计算的zkCluster集群列表
        restrictComputeZooKeeperClusterKeyList.clear();
        restrictComputeZooKeeperClusterKeyList.addAll(zooKeeperClusterKeyList);
        return zooKeeperClusterKeyToZooKeeperClusterMap;
    }

    private List<String> getZooKeeperClusterKeyList(Map<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap) {
        Collection<ZooKeeperCluster> zooKeeperClusterList = getZooKeeperClusterList(zooKeeperClusterKeyToZooKeeperClusterMap);
        List<String> zooKeeperClusterKeyList = new ArrayList<>();
        for (ZooKeeperCluster zooKeeperCluster : zooKeeperClusterList) {
            zooKeeperClusterKeyList.add(zooKeeperCluster.getZooKeeperClusterKey());
        }
        return zooKeeperClusterKeyList;
    }

    /**
     * 从数据库获取基础信息
     */
    public List<ZooKeeperCluster> getZooKeeperClusterList(Map<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap) {
        return new ArrayList<ZooKeeperCluster>(getZooKeeperClusterKeyToZooKeeperClusterMap().values());
    }

    private Map<String, ZooKeeperCluster> getZooKeeperClusterKeyToZooKeeperClusterMap() {
        Map<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap = new ConcurrentHashMap<String, ZooKeeperCluster>();
        List<ZooKeeperClusterDto> dtoList = zooKeeperClusterDtoService.getZooKeeperClusterInfoList();
        if (dtoList != null) {
            for (ZooKeeperClusterDto dto : dtoList) {
                if (tasksManagerId.equals(dto.getTasksManagerId())) {
                    ZooKeeperCluster zooKeeperCluster = new ZooKeeperCluster();
                    zooKeeperCluster.setZooKeeperClusterKey(dto.getZooKeeperClusterKey());
                    zooKeeperCluster.setAlias(dto.getAlias());
                    zooKeeperCluster.setZooKeeperAddressList(dto.getConnectString());
                    zooKeeperCluster.setDescription(dto.getDescription());
                    zooKeeperClusterKeyToZooKeeperClusterMap.put(dto.getZooKeeperClusterKey(), zooKeeperCluster);
                }
            }
        }
        return zooKeeperClusterKeyToZooKeeperClusterMap;
    }

    private void refreshZooKeeperClusterAndnameSpace(Map<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap) {
        // 对比旧的。
        // 1不包含的，关闭操作；
        // 2包含的，检查属性是否相同，如果相同，则直接赋值，否则，关闭旧的
        // 3其他直接关闭
        clearNotExistsAndNotEqualsZooKeeperCluster(zooKeeperClusterKeyToZooKeeperClusterMap);
        // 完善curatorFramework。如果没有，则新建
        initZooKeeper(zooKeeperClusterKeyToZooKeeperClusterMap);
        //
        List<String> onlinenameSpaceList = new ArrayList<String>();
        // 完善ZkCluster中的注册中心信息，关闭迁移了的域，新建迁移过来的域
        for (Map.Entry<String, ZooKeeperCluster> entry : zooKeeperClusterKeyToZooKeeperClusterMap.entrySet()) {
            String zooKeeperClusterKey = entry.getKey();
            ZooKeeperCluster zooKeeperCluster = entry.getValue();
            List<NameSpace> nameSpaceList = zooKeeperCluster.getNameSpaceList();
            //
            List<NameSpaceZooKeeperClusterMappingDto> nameSpaceZooKeeperClusterMappingDtoList = nameSpaceZooKeeperClusterMappingDtoService.getNameSpaceZooKeeperClusterMappingDtoList(zooKeeperClusterKey);
            closeNotExistsnameSpace(nameSpaceZooKeeperClusterMappingDtoList, nameSpaceList);
            initnameSpace(//
                    nameSpaceZooKeeperClusterMappingDtoList, nameSpaceList, zooKeeperCluster, onlinenameSpaceList //
            );
        }
        // 直接赋值新的
        BootStrap.zooKeeperClusterKeyToZooKeeperClusterMap = (LinkedHashMap<String, ZooKeeperCluster>) zooKeeperClusterKeyToZooKeeperClusterMap;
        log.info("new zkClusterMap is : {}", BootStrap.zooKeeperClusterKeyToZooKeeperClusterMap);
        this.onlinenameSpaceList = onlinenameSpaceList;
    }

    private void clearNotExistsAndNotEqualsZooKeeperCluster(Map<String, ZooKeeperCluster> newZooKeeperClusterKeyToZooKeeperClusterMap) {
        Iterator<Map.Entry<String, ZooKeeperCluster>> iterator = zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, ZooKeeperCluster> next = iterator.next();
            String zooKeeperClusterKey = next.getKey();
            ZooKeeperCluster zooKeeperCluster = next.getValue();
            //不包含则关闭
            if (!newZooKeeperClusterKeyToZooKeeperClusterMap.containsKey(zooKeeperClusterKey)) {
                iterator.remove();
                closeZooKeeperCluster(zooKeeperCluster);
            } else {
                ZooKeeperCluster newZooKeeperCluster = newZooKeeperClusterKeyToZooKeeperClusterMap.get(zooKeeperClusterKey);
                if (zooKeeperCluster.equals(newZooKeeperCluster)) {
                    newZooKeeperClusterKeyToZooKeeperClusterMap.put(zooKeeperClusterKey, zooKeeperCluster);
                } else if (zooKeeperCluster.equalsNoNeedReconnect(newZooKeeperCluster)) {
                    zooKeeperCluster.setDescription(newZooKeeperCluster.getDescription());
                    newZooKeeperClusterKeyToZooKeeperClusterMap.put(zooKeeperClusterKey, zooKeeperCluster);
                } else {
                    //不一致则关闭,但是这部分需要新建补充起来
                    iterator.remove();
                    closeZooKeeperCluster(zooKeeperCluster);
                }
            }
        }
    }

    /**
     * Close dashboardLeaderTreeCache, registryCenterClient, nameSpaceShardingListenerManager with this zkCluster
     */
    private void closeZooKeeperCluster(ZooKeeperCluster zooKeeperCluster) {
        try {
            List<NameSpace> nameSpaceList = zooKeeperCluster.getNameSpaceList();
            if (nameSpaceList != null) {
                for (NameSpace nameSpace1 : nameSpaceList) {
                    String nameSpace = nameSpace1.getNameSpace();
                    synchronized (getNameSpaceLock(nameSpace)) {
                        closenameSpaceClient(nameSpace);
                    }
                }
            }
            if (zooKeeperCluster.getConnectionStateListener() != null) {
                zooKeeperCluster.getConnectionStateListener().shutdownNowUntilTerminated();
                zooKeeperCluster.setConnectionStateListener(null);
            }
            if (zooKeeperCluster.getZooKeeper() != null && zooKeeperCluster.getZooKeeper().getCuratorFramework() != null) {
                zooKeeperCluster.getZooKeeper().getCuratorFramework().close();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void closenameSpaceClient(String nameSpace) {
        try {
            NameSpaceZooKeeperClient nameSpaceZooKeeperClient = nameSpaceTonameSpaceZooKeeperClientMap.remove(nameSpace);
            if (nameSpaceZooKeeperClient != null) {
                nameSpaceZooKeeperClient.close();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        try {
            TasksManager taskAllotManager = nameSpaceToTaskManagerMap.remove(nameSpace);
            if (taskAllotManager != null) {
                taskAllotManager.stopWithCuratorFramework();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }


    private void initZooKeeper(Map<String, ZooKeeperCluster> zooKeeperClusterKeyToZooKeeperClusterMap) {
        Iterator<Map.Entry<String, ZooKeeperCluster>> iterator = zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, ZooKeeperCluster> next = iterator.next();
            ZooKeeperCluster zooKeeperCluster = next.getValue();
            ZooKeeper zooKeeper = zooKeeperCluster.getZooKeeper();
            if (zooKeeper == null) {
                initZooKeeper(zooKeeperCluster);
                if (!zooKeeperCluster.isOffline()) {
                    //...
                }
            }
        }
    }

    private void initZooKeeper(final ZooKeeperCluster zooKeeperCluster) {
        String zooKeeperAddress = zooKeeperCluster.getZooKeeperAddressList();
        try {
            final CuratorFramework curatorFramework = zooKeeperRepository.connect(zooKeeperAddress, zooKeeperCluster.getDigest(), null);
            if (curatorFramework == null) {
                log.error("found an offline zkCluster, zkAddr is {}", zooKeeperAddress);
                zooKeeperCluster.setZooKeeper(null);
                zooKeeperCluster.setConnectionStateListener(null);
                zooKeeperCluster.setOffline(true);
            } else {
                BootStrapConnectionStateListener connectionStateListener = new BootStrapConnectionStateListener("zk-connectionListener-thread-for-zkCluster-" + zooKeeperCluster.getAlias()) {
                    @Override
                    public void stop() {
                        zooKeeperCluster.setOffline(true);
                    }

                    @Override
                    public void restart() {
                        try {
                            zooKeeperCluster.setOffline(false);
                        } catch (Exception e) {
                            log.error(e.getMessage(), e);
                        }
                    }
                };
                zooKeeperCluster.setZooKeeper(zooKeeperRepository.getZooKeeper(curatorFramework));
                zooKeeperCluster.setConnectionStateListener(connectionStateListener);
                zooKeeperCluster.setOffline(false);
                curatorFramework.getConnectionStateListenable().addListener(connectionStateListener);
            }
        } catch (Exception e) {
            log.error("found an offline zkCluster, zkAddr is {}", zooKeeperAddress);
            log.error(e.getMessage(), e);
            zooKeeperCluster.setZooKeeper(null);
            zooKeeperCluster.setConnectionStateListener(null);
            zooKeeperCluster.setOffline(true);
        }
    }


    private void closeNotExistsnameSpace(
            List<NameSpaceZooKeeperClusterMappingDto> nameSpaceZooKeeperClusterMappingDtoList,
            List<NameSpace> nameSpaceList) {
        if (nameSpaceList == null) {
            return;
        }
        Iterator<NameSpace> iterator = nameSpaceList.iterator();
        while (iterator.hasNext()) {
            NameSpace $nameSpace = iterator.next();
            String nameSpace = $nameSpace.getNameSpace();
            boolean include = false;
            if (nameSpaceZooKeeperClusterMappingDtoList != null) {
                for (NameSpaceZooKeeperClusterMappingDto dto : nameSpaceZooKeeperClusterMappingDtoList) {
                    if (nameSpace.equals(dto.getNameSpace())) {
                        include = true;
                        break;
                    }
                }
            }
            //关闭不存在的
            if (!include) {
                synchronized (getNameSpaceLock(nameSpace)) {
                    iterator.remove();
                    closenameSpaceClient(nameSpace);
                }
            }
        }
    }

    private void initnameSpace(//
                               List<NameSpaceZooKeeperClusterMappingDto> nameSpaceZooKeeperClusterMappingDtoList, List<NameSpace> nameSpaceList,//
                               ZooKeeperCluster zooKeeperCluster, List<String> onlinenameSpaceList//
    ) {//
        if (nameSpaceZooKeeperClusterMappingDtoList == null || zooKeeperCluster.isOffline()) {
            return;
        }//必须在线
        ZooKeeper zooKeeper = zooKeeperCluster.getZooKeeper();
        List<NameSpace> $nameSpaceList = new ArrayList<NameSpace>();
        try {
            for (NameSpaceZooKeeperClusterMappingDto dto : nameSpaceZooKeeperClusterMappingDtoList) {
                String nameSpace = dto.getNameSpace();
                // 如果这个标记为true，意味是新域
                boolean include = false;
                if (isnameSpaceNotIncludeInnameSpaceList(nameSpace, nameSpaceList)) {
                    // 对于新域，需要初始化节点信息
                    initnameSpaceZooKeeperNodeIfNecessary(nameSpace, zooKeeper);
                    include = true;
                }
                try {
                    if (include) {
                        NameSpace nameSpace1 = new NameSpace(zooKeeperCluster.getZooKeeperClusterKey(), zooKeeperCluster.getZooKeeperAddressList(), zooKeeperCluster.getDigest(), nameSpace);
                        $nameSpaceList.add(nameSpace1);
                        if (!onlinenameSpaceList.contains(nameSpace)) {
                            onlinenameSpaceList.add(nameSpace);
                        }
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        int oldSize = nameSpaceList != null ? nameSpaceList.size() : 0;
        if (oldSize != $nameSpaceList.size()) {
            log.info("Zkcluster [{}] nameSpace size change from {} to {}", zooKeeperCluster.getZooKeeperClusterKey(), oldSize, $nameSpaceList.size());
        }

        zooKeeperCluster.setNameSpaceList($nameSpaceList);
    }

    private boolean isnameSpaceNotIncludeInnameSpaceList(String nameSpace, List<NameSpace> nameSpaceList) {
        if (nameSpaceList == null || nameSpaceList.isEmpty()) {
            return true;
        }
        for (NameSpace $nameSpace : nameSpaceList) {
            if (nameSpace.equals($nameSpace.getNameSpace())) {
                return false;
            }
        }
        return true;
    }

    protected void initnameSpaceZooKeeperNodeIfNecessary(String nameSpace, ZooKeeper zooKeeper) {
        try {
            String taskIdsNodePath = TaskNodePath.tasks;
            if (!zooKeeper.checkExists(taskIdsNodePath)) {
                zooKeeper.create(taskIdsNodePath);
            }
            String executorIdsNodePath = "/" + nameSpace + ExecutorNodePath.getExecutorsNodePath();
            if (!zooKeeper.checkExists(executorIdsNodePath)) {
                zooKeeper.create(executorIdsNodePath);
            }
            String executorIdAndTaskIdsNodePath = "/" + nameSpace + "/executorIdAndTaskIds";
            if (!zooKeeper.checkExists(executorIdAndTaskIdsNodePath)) {
                zooKeeper.create(executorIdAndTaskIdsNodePath);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 创建或者移除nameSpaceShardingManager.
     */
    private void refreshnameSpaceToTaskManagerMap() {
        Iterator<Map.Entry<String, ZooKeeperCluster>> iterator = BootStrap.zooKeeperClusterKeyToZooKeeperClusterMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, ZooKeeperCluster> next = iterator.next();
            ZooKeeperCluster zooKeeperCluster = next.getValue();
            List<NameSpace> nameSpaceList = zooKeeperCluster.getNameSpaceList();
            if (nameSpaceList == null) {
                continue;
            }
            for (NameSpace $nameSpace : nameSpaceList) {
                String nameSpace = $nameSpace.getNameSpace();
                if (!nameSpaceToTaskManagerMap.containsKey(nameSpace)) {
                    if (restrictComputeZooKeeperClusterKeyList.contains($nameSpace.getZooKeeperClusterKey())) {
                        createnameSpaceTaskManager($nameSpace);
                    }
                } else {
                    TasksManager taskManager = nameSpaceToTaskManagerMap.get(nameSpace);
                    if (!restrictComputeZooKeeperClusterKeyList.contains($nameSpace.getZooKeeperClusterKey())) {
                        taskManager.stopWithCuratorFramework();
                        nameSpaceToTaskManagerMap.remove(nameSpace);
                    }
                }
            }
        }
    }

    private void createnameSpaceTaskManager(NameSpace nameSpaceConfiguration) {
        try {
            String nameSpace = nameSpaceConfiguration.getNameSpace();
            String digest = nameSpaceConfiguration.getDigest();
            CuratorFramework curatorFramework = zooKeeperRepository.connect(nameSpaceConfiguration.getZooKeeperAddressList(), digest, nameSpace);
            if (curatorFramework == null) {
                log.warn("fail to connect to zk during create nameSpaceShardingManager");
                return;
            }

            TasksManager taskManager = null;
            try {
                taskManager = new TasksManager(curatorFramework, nameSpace);
                taskManager.start();
                if (nameSpaceToTaskManagerMap.putIfAbsent(nameSpace, taskManager) != null) {
                    // 已经存在，则关闭当前的client
                    try {
                        taskManager.stopWithCuratorFramework();
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                    }
                } else {
                    log.info("Done starting nameSpaceShardingManager {}", nameSpace);
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                if (taskManager != null) {
                    try {
                        taskManager.stop();
                    } catch (Exception e2) {
                        log.error(e2.getMessage(), e2);
                    }
                }
                curatorFramework.close();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }


}